package com.bw.day0706;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author y3068
 * @Classname Test1
 * @Description TODO
 * @Date 2025/7/4 10:45
 * @Version V1.0
 */

public class age_test {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        tEnv.executeSql("create table dwd_ua( "+
                "log_id string," +
                "user_id string," +
                "item_id string," +
                "item_name string," +
                "behavior_type string," +
                "behavior_time string," +
                "search_keyword string," +
                "session_id string," +
                "stay_duration string," +
                "page_url string," +
                "category_id string," +
                "category_path string," +
                "keywords string," +
                "description string," +
                "price double," +
                "is_maternal string," +
                "create_time TIMESTAMP(3)," +
                "WATERMARK FOR create_time AS create_time - INTERVAL '5' SECOND " +
                ")"+ com.bw.utils.MyKafkaUtil.getKafkaDDL("dwd_ua","ua"));

//        Table table = tEnv.sqlQuery("select * from dwd_ua");
//        tEnv.toChangelogStream(table).print();

        Table table = tEnv.sqlQuery(
                "WITH a1 AS ( " +
                        "    SELECT " +
                        "        user_id, " +
                        "        item_id, " +
                        "        TUMBLE_START(create_time, INTERVAL '10' MINUTE) AS window_start, " +
                        "        behavior_time, " +
                        "        CASE behavior_type " +
                        "            WHEN 'purchase' THEN 1.0 " +
                        "            WHEN 'search'   THEN 0.8 " +
                        "            WHEN 'add_cart' THEN 0.7 " +
                        "            WHEN 'favorite' THEN 0.6 " +
                        "            ELSE 0.3 " +
                        "        END AS behavior_weight " +
                        "    FROM dwd_ua " +
                        "    GROUP BY user_id, item_id, TUMBLE(create_time, INTERVAL '10' MINUTE), behavior_time, behavior_type " +
                        "), " +
                        "a2 AS ( " +
                        "    SELECT " +
                        "        item_id, " +
                        "        TUMBLE_START(create_time, INTERVAL '10' MINUTE) AS window_start, " +
                        "        CASE " +
                        "            WHEN category_path LIKE '%孕妇装%' OR category_path LIKE '%防辐射服%' OR " +
                        "                 category_path LIKE '%孕妇裤%' OR category_path LIKE '%待产包%' OR " +
                        "                 category_path LIKE '%叶酸%' OR category_path LIKE '%钙铁锌%' THEN '孕早期（0-12 周）' " +
                        "            WHEN category_path LIKE '%孕妇装%' OR category_path LIKE '%孕妇护肤%' OR " +
                        "                 category_path LIKE '%孕妇奶粉%' OR category_path LIKE '%胎教产品%' THEN '孕中期（13-28 周）' " +
                        "            WHEN category_path LIKE '%婴儿床%' OR category_path LIKE '%奶瓶%' OR " +
                        "                 category_path LIKE '%吸奶器%' OR category_path LIKE '%乳头霜%' OR " +
                        "                 category_path LIKE '%月子餐%' THEN '孕晚期（29-40 周）' " +
                        "            WHEN category_path LIKE '%婴儿纸尿裤%' OR category_path LIKE '%婴儿湿巾%' OR " +
                        "                 category_path LIKE '%婴儿洗护%' OR category_path LIKE '%连身衣%' OR " +
                        "                 category_path LIKE '%婴儿帽%' THEN '0-3个月' " +
                        "            WHEN category_path LIKE '%婴儿辅食%' OR category_path LIKE '%婴儿奶瓶%' OR " +
                        "                 category_path LIKE '%婴儿安全座椅%' OR category_path LIKE '%安抚玩具%' OR " +
                        "                 category_path LIKE '%米粉%' OR category_path LIKE '%果泥%' THEN '3-6个月' " +
                        "            WHEN category_path LIKE '%婴儿学步鞋%' OR category_path LIKE '%儿童奶粉%' OR " +
                        "                 category_path LIKE '%DHA%' OR category_path LIKE '%益生菌%' OR " +
                        "                 category_path LIKE '%早教玩具%' THEN '6-12个月' " +
                        "            WHEN category_path LIKE '%幼儿服装%' OR category_path LIKE '%学步鞋%' OR " +
                        "                 category_path LIKE '%幼儿零食%' OR category_path LIKE '%幼儿安全座椅%' OR " +
                        "                 category_path LIKE '%调味品%' OR category_path LIKE '%学步车%' OR " +
                        "                 category_path LIKE '%连体裤%' THEN '12-24个月' " +
                        "            WHEN category_path LIKE '%益智%' OR category_path LIKE '%幼儿外套%' OR " +
                        "                 category_path LIKE '%幼儿帽子%' OR category_path LIKE '%绘本%' OR " +
                        "                 category_path LIKE '%学饮杯%' THEN '24-36个月' " +
                        "            WHEN category_path LIKE '%儿童T恤%' OR category_path LIKE '%儿童裤子%' OR " +
                        "                 category_path LIKE '%钙铁锌%' OR category_path LIKE '%拼图%' OR " +
                        "                 category_path LIKE '%积木%' OR category_path LIKE '%儿童读物%' THEN '3-6岁' " +
                        "            WHEN category_path LIKE '%运动鞋%' OR category_path LIKE '%文具%' OR " +
                        "                 category_path LIKE '%早教机%' OR category_path LIKE '%儿童零食%' OR " +
                        "                 category_path LIKE '%儿童护肤品%' THEN '6-12岁' " +
                        "            ELSE '未知' " +
                        "        END AS age_group, " +
                        "        1.0 as category_weight " +
                        "    FROM dwd_ua " +
                        "    GROUP BY item_id, TUMBLE(create_time, INTERVAL '10' MINUTE), category_path " +
                        "), " +
                        "a3 AS ( " +
                        "    SELECT " +
                        "        a1.user_id, " +
                        "        a2.age_group, " +
                        "        a1.window_start, " +
                        "        ROUND(SUM(a1.behavior_weight * a2.category_weight), 2) AS total_score, " +
                        "        MAX(a1.behavior_time) as latest_time " +
                        "    FROM a1 " +
                        "    JOIN a2 ON a1.item_id = a2.item_id AND a1.window_start = a2.window_start " +
                        "    GROUP BY a1.user_id, a2.age_group, a1.window_start " +
                        "), " +
                        "max_scores AS ( " +
                        "    SELECT " +
                        "        user_id, " +
                        "        MAX(total_score) as max_score " +
                        "    FROM a3 " +
                        "    GROUP BY user_id " +
                        "), " +
                        "latest_records AS ( " +
                        "    SELECT " +
                        "        user_id, " +
                        "        age_group, " +
                        "        window_start, " +
                        "        latest_time " +
                        "    FROM ( " +
                        "        SELECT " +
                        "            user_id, " +
                        "            age_group, " +
                        "            window_start, " +
                        "            latest_time, " +
                        "            ROW_NUMBER() OVER (PARTITION BY user_id ORDER BY latest_time DESC) as rn " +
                        "        FROM a3 " +
                        "    ) WHERE rn = 1 " +
                        ") " +
                        "SELECT " +
                        "    a3.user_id, " +
                        "    a3.window_start, " +
                        "    COALESCE( " +
                        "        MAX(CASE WHEN a3.total_score * 2 < ms.max_score THEN a3.age_group END), " +
                        "        lr.age_group " +
                        "    ) AS age_group " +
                        "FROM a3 " +
                        "JOIN max_scores ms ON a3.user_id = ms.user_id " +
                        "LEFT JOIN latest_records lr ON a3.user_id = lr.user_id " +
                        "GROUP BY a3.user_id, a3.window_start, lr.age_group"
        );

        tEnv.toChangelogStream(table).print();

        env.execute();
    }
}
